package ru.ifmo.croak.impl.stress.query.team1;

import ru.ifmo.croak.api.Message;
import ru.ifmo.croak.api.MessageQuery;
import ru.ifmo.croak.api.QueryRequest;
import ru.ifmo.croak.api.QueryService;

import java.util.Random;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author Roman Kolganov
 *         26.05.11
 */
public class QueryRequester implements Runnable {

    private final Random rand = new Random();
    private final QueryService query;
    private final AtomicLong sent = new AtomicLong(0);
    private final AtomicLong queryReceived = new AtomicLong(0);
    private final AtomicLong symbolsReceived = new AtomicLong(0);
    private final AtomicLong aborted = new AtomicLong(0);
    private final QueryRequestGenerator generator = new QueryRequestGenerator();
    private final Executor executor = new ThreadPoolExecutor(2, 10, 5,
            TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(100),
            new ThreadPoolExecutor.CallerRunsPolicy());

    public QueryRequester(QueryService query) {
        this.query = query;
    }

    public long sent() {
        return sent.get();
    }

    public long queryReceived() {
        return queryReceived.get();
    }

    public long symbolsReceived() {
        return symbolsReceived.get();
    }

    public long aborted() {
        return aborted.get();
    }

    public void run() {
        while (!Thread.interrupted()) {
            final QueryRequest request = generator.nextQueryRequest(rand);
            sent.incrementAndGet();
            executor.execute(new Runnable() {
                public void run() {
                    try {
                        MessageQuery messageQuery = query.createQuery(request);
                        queryReceived.incrementAndGet();
                        messageQuery.setMessageListener(
                                new CountingMessageListener(symbolsReceived,
                                        generator));
                        for (Message message : messageQuery.getMessages()) {
                            symbolsReceived.addAndGet(message.getText().length());
                        }
                    } catch (RuntimeException e) {
                        aborted.incrementAndGet();
                    }
                }
            });
        }
    }

}
